package org.apache.nifi.processors.huawei.smn;

import com.huaweicloud.sdk.smn.v2.SmnClient;
import com.huaweicloud.sdk.smn.v2.model.PublishMessageRequest;
import com.huaweicloud.sdk.smn.v2.model.PublishMessageRequestBody;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.*;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.huawei.abstractprocessor.AbstractSMNProcessor;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;

@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"HuaweiCloud", "SMN",  "Publish", "Message"})
@CapabilityDescription("This API is used to publish messages to a topic. After the message ID is returned, " +
        "the message has been saved and is to be pushed to the subscribers of the topic.")
public class PublishSMNMessage extends AbstractSMNProcessor {

    public static final PropertyDescriptor SMN_PROJECT_ID = new PropertyDescriptor.Builder()
            .name("project_id")
            .displayName("Project ID")
            .description("Project ID")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .required(true)
            .build();

    public static final PropertyDescriptor SMN_TOPIC_URN = new PropertyDescriptor.Builder()
            .name("topic_urn")
            .displayName("Topic URN")
            .description("The unique resource identifier of the Topic")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .required(true)
            .build();

    public static final PropertyDescriptor SMN_SUBJECT = new PropertyDescriptor.Builder()
            .name("subject")
            .displayName("Subject")
            .description("Specifies the message subject, which is used as the email subject when you publish email messages. " +
                    "The length of the subject cannot exceed 512 bytes.")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .build();

    public static final PropertyDescriptor SMN_MESSAGE = new PropertyDescriptor.Builder()
            .name("message")
            .displayName("Message")
            .description("Specifies the message content. The message content must be UTF-8-coded and can be no more than 256 KB.")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor SMN_MESSAGE_STRUCTURE = new PropertyDescriptor.Builder()
            .name("message_structure")
            .displayName("Message Structure")
            .description("Specifies the message structure, which contains JSON strings. " +
                    "email, sms, http, https, dms, function stage, HMS, APNS, or APNS_SANDBOX are supported.")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor SMN_MESSAGE_TEMPLATE_NAME = new PropertyDescriptor.Builder()
            .name("message_template_name")
            .displayName("Message Template Name")
            .description("Specifies the message template name. To obtain the name, see Querying Message Templates.")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final PropertyDescriptor SMN_TAGS = new PropertyDescriptor.Builder()
            .name("tags")
            .displayName("Tags")
            .description("Specifies the dictionary consisting of variable parameters and values. " +
                    "This parameter is mandatory when you use a message template to publish messages.")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .dependsOn(SMN_MESSAGE_TEMPLATE_NAME)
            .build();

    public static final PropertyDescriptor SMN_TIME_TO_LIVE = new PropertyDescriptor.Builder()
            .name("time_to_live")
            .displayName("Time to Live")
            .description("Specifies the maximum retention period of a message in SMN. " +
                    "After the retention period expires, SMN does not send this message.")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();

    public static final int MAX_SIZE = 256 * 1024;

    private static final List<PropertyDescriptor> properties = List.of(ACCESS_KEY, SECRET_KEY, SMN_PROJECT_ID, SMN_TOPIC_URN, SMN_REGION,
            SMN_SUBJECT, SMN_MESSAGE,
            SMN_MESSAGE_STRUCTURE, SMN_MESSAGE_TEMPLATE_NAME,
            SMN_TAGS, SMN_TIME_TO_LIVE);

    @Override
    public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return properties;
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) {

        FlowFile flowFile = session.get();
        if (flowFile == null) {

            return;
        }

        if (flowFile.getSize() > MAX_SIZE) {

            getLogger().error("Cannot publish {} to SMN because its size exceeds Huawei SMN's limit of 256KB; routing to failure", new Object[]{flowFile});
            session.transfer(flowFile, REL_FAILURE);

            return;
        }

        try {

            String topicUrn = context.getProperty(SMN_TOPIC_URN).evaluateAttributeExpressions(flowFile).getValue();
            String subject = context.getProperty(SMN_SUBJECT).evaluateAttributeExpressions(flowFile).getValue();
            String message = context.getProperty(SMN_MESSAGE).evaluateAttributeExpressions(flowFile).getValue();
            String messageStructure = context.getProperty(SMN_MESSAGE_STRUCTURE).evaluateAttributeExpressions(flowFile).getValue();
            String messageTemplateName = context.getProperty(SMN_MESSAGE_TEMPLATE_NAME).evaluateAttributeExpressions(flowFile).getValue();
            String tags = context.getProperty(SMN_TAGS).evaluateAttributeExpressions(flowFile).getValue();
            String timeToLive = context.getProperty(SMN_TIME_TO_LIVE).evaluateAttributeExpressions(flowFile).getValue();

            PublishMessageRequest request = new PublishMessageRequest();
            request.setTopicUrn(topicUrn);

            PublishMessageRequestBody body = new PublishMessageRequestBody();
            body.setSubject(subject);
            body.setMessage(message);
            body.setMessageStructure(messageStructure);
            body.setMessageTemplateName(messageTemplateName);
            body.setTimeToLive(timeToLive);

            if (!StringUtils.isEmpty(tags)) {

                Map<String, String> tagsMap = new ObjectMapper().readValue(tags, HashMap.class);
                body.setTags(tagsMap);
            }

            request.withBody(body);
            SmnClient smnClient = getSmnClient();

            flowFile = session.putAllAttributes(flowFile, createSmnAttributes(body));
            smnClient.publishMessage(request);

            session.transfer(flowFile, REL_SUCCESS);
            session.getProvenanceReporter().send(flowFile, topicUrn);

            getLogger().info("Successfully published notification for {}", new Object[]{flowFile});

        } catch (Exception e) {

            getLogger().error("Failed to publish message for {} due to {}", new Object[]{flowFile, e});

            flowFile = session.penalize(flowFile);
            session.transfer(flowFile, REL_FAILURE);
        }
    }

    private Map<String, String> createSmnAttributes(PublishMessageRequestBody body) {

        Map<String, String> attributes = new HashMap<>();

        if (!StringUtils.isEmpty(body.getSubject())) {

            attributes.put(SMN_SUBJECT.getName(), body.getSubject());
        }

        if (!StringUtils.isEmpty(body.getMessage())) {

            attributes.put(SMN_MESSAGE.getName(), body.getMessage());
        }

        if (!StringUtils.isEmpty(body.getMessageStructure())) {

            attributes.put(SMN_MESSAGE_STRUCTURE.getName(), body.getMessageStructure());
        }

        if (!StringUtils.isEmpty(body.getMessageTemplateName())) {

            attributes.put(SMN_MESSAGE_TEMPLATE_NAME.getName(), body.getMessageTemplateName());
        }

        if (!StringUtils.isEmpty(body.getTimeToLive())) {

            attributes.put(SMN_TIME_TO_LIVE.getName(), body.getTimeToLive());
        }

        return attributes;
    }
}
